package com.tangsm.spring.boot.degradation.aspect;

import com.google.common.util.concurrent.RateLimiter;
import com.tangsm.spring.boot.degradation.annotation.ResourceDegradation;
import com.tangsm.spring.boot.degradation.util.SpringContextUtil;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@Aspect
public class ResourceDegradationAspect {
    private static final Logger log = LoggerFactory.getLogger(ResourceDegradationAspect.class);


    @Around(value = "@annotation(resourceDegradation)")
    public Object resouceDowngrade(ProceedingJoinPoint pjp, ResourceDegradation resourceDegradation)
            throws Throwable {

        return this.downgradeWithRateLimiter(pjp, resourceDegradation);
    }

    //---------利用guava的ratelimter进行限流降级------------------------------------//

    private static final Map<String, RateLimiter> rateLimiterMap = new ConcurrentHashMap<>();

    /**
     * 存放初始阈值
     */
    private Map<String, Integer> orginMaxThresholdMap = new ConcurrentHashMap<>();

    private Object downgradeWithRateLimiter(ProceedingJoinPoint pjp, ResourceDegradation resourceDegradation) throws Throwable {

        String signature = pjp.getSignature().toLongString();
        int maxThreshold = resourceDegradation.maxThreshold();

        log.info("maxThreshold-->{}", maxThreshold);
        //permitsPerSecond为每秒往令牌里放入几个令牌
        RateLimiter rateLimiter;
        if (!rateLimiterMap.containsKey(signature)) {
            rateLimiter = RateLimiter.create(maxThreshold);
            rateLimiterMap.put(signature, rateLimiter);
        }

        rateLimiter = this.resetRateLimiterIfMaxThresholdChange(signature, maxThreshold);

        boolean isAllowPass = rateLimiter.tryAcquire();
        if (isAllowPass) {
            return pjp.proceed();
        }

        return this.fallback(pjp, resourceDegradation);

    }

    /**
     * 阈值发生动态切换，此时则需要变更限流器阈值
     *
     * @param signature
     * @param maxThreshold
     * @return
     */
    private RateLimiter resetRateLimiterIfMaxThresholdChange(String signature, int maxThreshold) {

        int orginMaxThreshold;
        RateLimiter rateLimiter = rateLimiterMap.get(signature);
        if (orginMaxThresholdMap.containsKey(signature)) {
            orginMaxThreshold = orginMaxThresholdMap.get(signature);
            //说明发生阈值发生动态切换，此时则需要变更限流器
            if (orginMaxThreshold != maxThreshold) {
                log.info("signature:{}-->orginMaxThreshold is :{},currentMaxThreshold is :{},rateLimiter maxThreshold will change to -->{}"
                        , signature, orginMaxThreshold, maxThreshold, maxThreshold);
                rateLimiter.setRate(maxThreshold);
            }
        } else {
            orginMaxThresholdMap.put(signature, maxThreshold);
        }
        return rateLimiter;
    }

    //----------普通计数器进行限流------------------------------------//
    private Map<String, AtomicInteger> counter = new ConcurrentHashMap<>();

    private Object downgrade(ProceedingJoinPoint pjp, ResourceDegradation resourceDegradation) throws Throwable {

        String signature = pjp.getSignature().toLongString();
        int maxThreshold = resourceDegradation.maxThreshold();
        log.info("maxThreshold-->{}", maxThreshold);
        if (counter.containsKey(signature)) {
            if (counter.get(signature).incrementAndGet() > maxThreshold) {
                log.info("signature:{},current requestCount:{},maxThreshold:{}", signature, counter.get(signature).get(), maxThreshold);
                return this.fallback(pjp, resourceDegradation);
            }
        } else {
            counter.put(signature, new AtomicInteger(1));
        }

        //极端情况下，当maxThreshold为0
        if (counter.get(signature).get() > maxThreshold) {
            log.info("signature:{},current requestCount:{},maxThreshold:{}", signature, counter.get(signature).get(), maxThreshold);
            return this.fallback(pjp, resourceDegradation);
        }
        Object result;
        try {
            result = pjp.proceed();
        } catch (Throwable throwable) {
            log.error(throwable.getMessage());
            throw throwable;
        } finally {
            counter.get(signature).decrementAndGet();
        }
        return result;
    }

    /**
     * 限流回退逻辑
     *
     * @param pjp 进行连接点
     * @param resourceDegradation
     * @return
     */
    private Object fallback(ProceedingJoinPoint pjp, ResourceDegradation resourceDegradation) {

        String methodName = resourceDegradation.fallbackMethod();
        if (methodName == null || "".equals(methodName)) {
            methodName = pjp.getSignature().getName();
        }
        try {
            MethodSignature methodSignature = (MethodSignature) pjp.getSignature();
            Class<?> aClass = resourceDegradation.fallbackClass();
            Method method = SpringContextUtil.getBean(aClass).getClass().getMethod(methodName, methodSignature.getParameterTypes());
            log.info("enter fallbackMethod:{}", method);
            return method.invoke(SpringContextUtil.getBean(aClass), pjp.getArgs());
        } catch (NoSuchMethodException | IllegalAccessException | IllegalArgumentException |
                 InvocationTargetException e) {
            log.error("fallback error:" + e.getMessage(), e);
        }

        return null;
    }


}
